package com.sunzm.flink.datastream.scala.time

import java.time.Duration

import com.alibaba.fastjson.JSON
import org.apache.commons.lang3.StringUtils
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
import org.apache.flink.streaming.api.operators.source.TimestampsAndWatermarksContext
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 *
 * ScalaEventTimeDemo
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-06-22 20:49
 */
object ScalaEventTimeDemo {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {

    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    //Flink1.11之前，默认的时间语义是：处理时间，从Flink1.11开始，默认的时间语义就是事件时间
    //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //获取数据
    //{"callRecordId":"call-202106260001","startTime":1624685550000,"endTime":1624685808000,
    // "caller":"18888885656","callee":"10086","direction":0,"dateTime":1624685810000,
    // "messageType":"callRecord"}
    val dataStream: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)
      .filter(line => {
        var res = false

        try {
          if (StringUtils.isNotBlank(line)) {
            val jSONObject = JSON.parseObject(line)

            val dateTime = jSONObject.getLongValue("dateTime")
            val callRecordId = jSONObject.getString("callRecordId")

            if (dateTime > 0 && StringUtils.isNotBlank(callRecordId)) {
              res = true
            }
          }
        } catch {
          case e: Throwable => {
            logger.warn(s"JSON解析异常: ${line}", e)
          }
        }

        res
      })

    //定义WatermarkStrategy
    val watermarkStrategy = WatermarkStrategy.forBoundedOutOfOrderness[String](Duration.ofSeconds(5))
      .withTimestampAssigner(new SerializableTimestampAssigner[String] {
        override def extractTimestamp(element: String, recordTimestamp: Long): Long = {

          val jSONObject = JSON.parseObject(element)

          jSONObject.getLongValue("dateTime")

        }
      })

    //需要注意并行度，必须每个分区的WaterWark都到达定时器触发的时间，定时器才会被触发

    val resultDS = dataStream
      //指定WatermarkStrategy
      .assignTimestampsAndWatermarks(watermarkStrategy)
      .keyBy(line => {
        val jSONObject = JSON.parseObject(line)

        jSONObject.getString("callRecordId")
      })
      //开始处理数据
      .process(new KeyedProcessFunction[String, String, String] {

        /**
         * 用来保存数据的状态
         */
        private var dataState: ValueState[String] = _

        /**
         * 用来保存定时器时间的状态
         */
        private var timerState: ValueState[Long] = _

        override def open(parameters: Configuration): Unit = {
          dataState = getRuntimeContext.getState(
            new ValueStateDescriptor[String]("dataState", classOf[String])
          )
          timerState = getRuntimeContext.getState(
            new ValueStateDescriptor[Long]("timerState", classOf[Long])
          )
        }

        override def processElement(value: String,
                                    ctx: KeyedProcessFunction[String, String, String]#Context,
                                    out: Collector[String]): Unit = {

          dataState.update(value)

          //注册一个基于事件时间的20秒定时器
          //ctx.timestamp这个就是事件时间
          logger.info(s"ctx.timestamp: ${ctx.timestamp}")
          logger.info(s"currentWatermark: ${ctx.timerService().currentWatermark()}")
          val timerTimeMillis = ctx.timestamp + Time.seconds(20).toMilliseconds
          logger.info(s"timerTimeMillis: ${timerTimeMillis}")

          ctx.timerService().registerEventTimeTimer(timerTimeMillis)
          //把触发定时器执行的时间戳保存到状态中
          timerState.update(timerTimeMillis)

        }

        override def onTimer(timestamp: Long,
                             ctx: KeyedProcessFunction[String, String, String]#OnTimerContext,
                             out: Collector[String]): Unit = {
          val data = dataState.value()

          if (StringUtils.isNotBlank(data)) {
            //输出数据
            out.collect(data)

            //清空状态
            dataState.clear()
          }

          //把记录定时器时间的状态清空
          timerState.clear()

        }

        override def close(): Unit = {
          if(dataState != null){
            dataState.clear()
          }
          if(timerState != null){
            timerState.clear()
          }
        }
      })

    resultDS.print()

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }
}
